package Join;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, ReduceJoinBean> {
    Text k = new Text();
    ReduceJoinBean v = new ReduceJoinBean();
    String fileName;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        FileSplit fs = (FileSplit) context.getInputSplit();
        fileName = fs.getPath().getName();
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        String[] datas = value.toString().split(",");
        if ("orders.txt".equals(fileName)) {
            k.set(datas[2]);
            v.set(datas[0], datas[1], datas[2], datas[3], "", "", "", fileName);
            context.write(k, v);
            System.out.println(v+v.getFileName());
        } else {
            k.set(datas[0]);
            v.set("", "", datas[0], "", datas[1], datas[2], datas[3], fileName);
            context.write(k, v);
            System.out.println(v+v.getFileName());
        }
    }
}
